/*
 * Proto Streamer
 * Copyright (C) 2009 John Pritchard.
 *
 * This program is free software; you can redistribute it and/or
 * modify it under the terms of the GNU General Public License as
 * published by the Free Software Foundation; either version 2 of the
 * License, or (at your option) any later version.
 * 
 * This program is distributed in the hope that it will be useful, but
 * WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
 * 02110-1301 USA.
 */
package p.s.q;

import p.s.m.Consumer;
import p.s.m.Producer;
import com.google.protobuf.Message;
import alto.io.u.Lobjmap;
import alto.io.u.Objmap;
import java.lang.reflect.InvocationTargetException;

/**
 * A message queue from which consumer / downlinks read unread
 * messages posted by producer / uplinks.
 * 
 * A session is identified by an identity.  A typical identity would
 * be a {@link java.net.URI}, but this package imposes no restrictions
 * on the class of its identities.
 * 
 * @author jdp
 */
public class Queue
    extends p.s.m.Base
{
    /**
     * Queue constructor used by session manager.
     */
    public final static class Constructor
        extends Object
    {
        public final Class<? extends Queue> clas;

        public final java.lang.reflect.Constructor<? extends Queue> ctor;

        public Constructor(Class<? extends Queue> clas)
            throws NoSuchMethodException
        {
            super();
            assert null != clas;
            this.clas = clas;
            this.ctor = clas.getConstructor(Object.class);
            assert null != this.ctor;
        }

        public Queue newInstance(Object id)
            throws InstantiationException, IllegalAccessException,
                   IllegalArgumentException, InvocationTargetException
        {
            return this.ctor.newInstance(id);
        }
    }


    protected volatile Message[] buffer;

    protected volatile long last, timeout = (5L * Minutes);


    public Queue(Object id){
        super(id);
    }


    protected Message update(Consumer c){
        this.lock.lockReadEnter();
        try {
            Message[] buffer = this.buffer;
            if (null != buffer){
                int p = c.getServicePointer(this.identity,this.version);
                if (p < buffer.length){
                    Message m = buffer[p++];
                    c.setServicePointer(this.identity,this.version,p);
                    return m;
                }
            }
            return null;
        }
        finally {
            this.lock.lockReadExit();
        }
    }
    protected void update(Message m){
        this.lock.lockWriteEnter();
        try {
            this.buffer = Add(this.buffer,m);
        }
        finally {
            this.lock.lockWriteExit();
        }
        synchronized(this.monitor){
            this.monitor.notifyAll();
        }
    }
    public void update(Producer p, Message m){
        if (null != m)
            this.update(m);
        else
            throw new IllegalArgumentException("Null message");
    }
    public void clear(Producer p){
        this.lock.lockWriteEnter();
        try {
            this.buffer = null;
            this.version++;
        }
        finally {
            this.lock.lockWriteExit();
        }
    }
    /**
     * Mark activity to prevent expiration
     */
    public Queue mark(){
        this.last = System.currentTimeMillis();
        return this;
    }
    /**
     * @param timeout Number of milliseconds since last activity for
     * expiration.  Zero for none, expire with no members.  Default
     * five minutes.
     */
    public void setTimeout(long timeout){
        if (-1 < timeout)
            this.timeout = timeout;
        else
            throw new IllegalArgumentException(String.valueOf(timeout));
    }
    public boolean hasExpired(){
        if (this.isEmpty()){
            long timeout = this.timeout;
            if (0L < timeout){
                long exp = (System.currentTimeMillis()- timeout);
                return (this.last < exp);
            }
            else
                return true;
        }
        else
            return false;
    }
    public boolean isEmpty(){
        return (this.producers.isEmpty() && this.consumers.isEmpty());
    }
    public boolean isNotEmpty(){
        return (this.producers.isNotEmpty() || this.consumers.isNotEmpty());
    }
    protected final static Message[] Add(Message[] list, Message item){
        if (null == list)
            return new Message[]{item};
        else {
            int len = list.length;
            Message[] copy = new Message[len+1];
            System.arraycopy(list,0,copy,0,len);
            copy[len] = item;
            return copy;
        }
    }
    public void init(){
    }
    public void destroy(){
    }
}
